package org.peoplesearch.idfilter;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

public class FilterEngine {

    private Collection<Filter> filters;
    private Iterable<Long> ids;
    private String directory;
    private Boolean batch;
    private Integer batchSize;

    final static Logger logger = LoggerFactory.getLogger(FilterEngine.class);

    public FilterEngine(Collection<Filter> filters, Iterable<Long> ids, String directory, Boolean batch, Integer batchSize) {
        this.filters = filters;
        this.ids = ids;
        this.directory = directory;
        this.batch = batch;
        this.batchSize = batchSize;
    }

    public static void main(String[] args) throws RuntimeException, InterruptedException, ExecutionException, IOException {

        ApplicationContext ctx = new ClassPathXmlApplicationContext(
                "executor.xml",
                "filterEngine.xml");

        FilterEngine filterEngine = (FilterEngine) ctx.getBean("filterEngine");
        filterEngine.filter();

    }

    private void filter() throws InterruptedException, ExecutionException, IOException {

        // Create for every filter result list

        final Map<Filter, Set<Long>> result = new HashMap<Filter, Set<Long>>();

        for (Filter filter : filters) {
            result.put(filter, Collections.synchronizedSet(new HashSet<Long>()));
        }

        // Get all id's in memory

        final Queue<Long> allId = new ConcurrentLinkedQueue<Long>();

        for (Long id : ids) {
            allId.add(id);
        }

        // Casual store

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    logger.debug("Started storing in the files");
                    storeIds(result, Calendar.getInstance().getTimeInMillis() + "");
                    logger.debug("Completed storing in the files");
                } catch (Exception e) {
                    e.printStackTrace();
                    System.exit(-1);
                }
            }
        }, 20, 50, TimeUnit.SECONDS);

        // Create executor to add to result


        int threads = 50;

        ExecutorService executor = Executors.newFixedThreadPool(threads);
        List<Future> futures = new LinkedList<Future>();

        for (int i = 0; i < 50; i++) {
            futures.add(executor.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    if (batch) {
                        Collection<Long> ids = new LinkedList<Long>();
                        int i = 0;
                        Long id = allId.poll();
                        while (id != null) {
                            if (ids.size() >= batchSize) {
                                storeResult(ids);
                                ids.clear();
                            }
                            ids.add(id);
                            id = allId.poll();
                        }
                        storeResult(ids);

                    } else {
                        Long id = allId.poll();
                        while (id != null) {
                            for (Filter filter : result.keySet()) {
                                if (filter.match(id)) {
                                    result.get(filter).add(id);
                                }
                            }
                            id = allId.poll();
                        }
                    }
                    return null;
                }

                private void storeResult(Collection<Long> ids) {
                    for (Filter filter : result.keySet()) {
                        Map<Long, Boolean> matches = filter.match(ids);
                        for (Long matched: matches.keySet()){
                            if (matches.get(matched)){
                                result.get(filter).add(matched);
                            }
                        }
                    }
                }
            }));
        }

        // Wait for processing;

        for (Future future : futures) {
            future.get();
        }

        // Store all ids

        storeIds(result, "");

    }

    private void storeIds(Map<Filter, Set<Long>> result, String suffix) throws IOException {
        for (Filter filter : result.keySet()) {
            BufferedWriter filterFile = new BufferedWriter(new FileWriter(new File(directory + "/" + filter.getName() + "_" + suffix)));
            filterFile.write(result.get(filter).size() + "\n");
            for (Long id : result.get(filter)) {
                filterFile.write(id + "\n");
            }
            filterFile.close();
        }
    }

}
